package org.apache.zookeeper.server.paxos2;

import java.io.*;
import java.nio.ByteBuffer;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;

/**
 * Created by IntelliJ IDEA.
 * User: iodine
 * Date: Oct 22, 2011
 * Time: 2:22:27 PM
 * To change this template use File | Settings | File Templates.
 */
public class LogSystem {

    private final String path;
    private final long size;

    private long begin = 0;
    private long end = 0;

    private LinkedHashMap<String, FileMeta> meta = new LinkedHashMap<String, FileMeta>();

    private RandomAccessFile f;

    private static class FileMeta {
        public long begin;
        public long end;

        public FileMeta(long begin, long end) {
            this.begin = begin;
            this.end = end;
        }
    }

    public LogSystem(String path, long size) throws IOException {
        this.path = path;
        this.size = size;
        File metaFile = new File(path + ".meta");
        if (metaFile.exists())
            deserializeMeta();
        try {
            f = new RandomAccessFile(path, "rw");
        }
        catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void format() throws IOException {
        byte[] tmp = new byte[4096];
        f.seek(0);
        for (long i = 0; i < size / tmp.length; i++)
            f.write(tmp);
        meta.clear();
        begin = 0;
        end = 0;
    }

    private void serializeMeta() throws IOException {
        File metaFile = new File(path + ".meta");
        DataOutputStream dos = new DataOutputStream(new FileOutputStream(metaFile));
        dos.writeLong(begin);
        dos.writeLong(end);
        dos.writeInt(meta.size());
        for (String fileName : meta.keySet()) {
            dos.writeUTF(fileName);
            FileMeta tmp = meta.get(fileName);
            dos.writeLong(tmp.begin);
            dos.writeLong(tmp.end);
        }

    }

    private void deserializeMeta() throws IOException {
        File metaFile = new File(path + ".meta");
        DataInputStream din = new DataInputStream(new FileInputStream(metaFile));
        begin = din.readLong();
        end = din.readLong();

        int size = din.readInt();
        meta.clear();
        for (int i = 0; i < size; i++) {
            String fileName = din.readUTF();
            FileMeta tmp = new FileMeta(din.readLong(), din.readLong());
            meta.put(fileName, tmp);
        }
    }

    public synchronized OutputStream newLog(String fileName) throws IOException {
        return new LogOutputStream(fileName);
    }

    public synchronized InputStream readLog(String fileName) throws IOException {
        return new LogInputStream(fileName);
    }

    public List<String> listFiles() {
        LinkedList<String> ret = new LinkedList<String>();
        for (String fileName : meta.keySet())
            ret.addLast(fileName);
        return ret;
    }

    public synchronized void remove(String fileName) throws IOException {
        FileMeta fileMeta = meta.get(fileName);
        if (fileMeta == null)
            throw new IOException("File not found " + fileName);
        if (fileMeta.begin != fileMeta.begin)
            throw new IOException("Not the first log file, can only remove the first one " + fileName);
        begin = fileMeta.end;
        meta.remove(fileName);
        serializeMeta();
    }

    public class LogOutputStream extends OutputStream {
        private RandomAccessFile out;
        private final String fileName;
        private final FileMeta fileMeta;

        public LogOutputStream(String fileName) throws IOException {

            if (meta.containsKey(fileName))
                throw new IOException(fileName + " already exists");
            this.fileName = fileName;
            fileMeta = new FileMeta(end, end);
            meta.put(fileName, fileMeta);
            out = new RandomAccessFile(path, "rw");
            out.seek(end);
        }

        public void write(int b) throws IOException {
            throw new IOException("Not implemented");
        }

        @Override
        public void write(byte[] data) throws IOException {
            write(data, 0, data.length);
        }

        public void write(byte data[], int off, int len) throws IOException {
            if (end - begin + len > size)
                throw new IOException("Log space full, Need garbage collection");
            if (end + len <= size)
                out.write(data, off, len);
            else {
                int remaining = (int) (size - end);
                out.write(data, off, remaining);
                out.seek(0);
                out.write(data, off + remaining, len - remaining);
            }
            end = (end + len) % size;
            fileMeta.end = end;
        }

        @Override
        public void flush() throws IOException {
            serializeMeta();
            out.getFD().sync();
        }

        @Override
        public void close() throws IOException {
            serializeMeta();
            out.getFD().sync();
        }
    }

    public class LogInputStream extends InputStream {

        private final String fileName;
        private final FileMeta fileMeta;
        private RandomAccessFile in;

        public LogInputStream(String fileName) throws IOException {
            this.fileName = fileName;
            this.fileMeta = meta.get(fileName);
            System.out.println(fileName + " begin=" + fileMeta.begin + " end=" + fileMeta.end);
            if (fileMeta == null)
                throw new IOException("File not found " + fileName);
            in = new RandomAccessFile(path, "r");
            in.seek(fileMeta.begin);
        }

        @Override
        public int read() throws IOException {
            throw new IOException("Not implemented");
        }

        @Override
        public int read(byte b[], int off, int len) throws IOException {
            if (b == null) {
                throw new NullPointerException();
            } else if (off < 0 || len < 0 || len > b.length - off) {
                throw new IndexOutOfBoundsException();
            } else if (len == 0) {
                return 0;
            }

            int target = len;
            if (remaining() < target)
                target = (int) remaining();
            if (target == 0)
                return 0;

            if (fileMeta.end > fileMeta.begin) {
                in.readFully(b, off, target);
                return target;
            } else {
                if (size - in.getFilePointer() >= target) {
                    in.readFully(b, off, target);
                    return target;
                } else {
                    int tmp = (int) (size - in.getFilePointer());
                    in.readFully(b, off, tmp);
                    in.seek(0);
                    in.readFully(b, off + tmp, target - tmp);
                    return target;
                }

            }

        }

        @Override
        public long skip(long n) throws IOException {
            int target = (int) n;
            if (remaining() < target)
                target = (int) remaining();
            if (target == 0)
                return 0;

            if (fileMeta.end > fileMeta.begin) {
                in.skipBytes(target);
                return target;
            } else {
                if (size - in.getFilePointer() >= target) {
                    in.skipBytes(target);
                    return target;
                } else {
                    int tmp = (int) (size - in.getFilePointer());
                    in.seek(0);
                    in.skipBytes(target - tmp);
                    return target;
                }

            }
        }

        private long remaining() throws IOException {
            if (fileMeta.end > fileMeta.begin)
                return fileMeta.end - in.getFilePointer();
            else{
                if(in.getFilePointer() > fileMeta.end)
                    return fileMeta.end + size - in.getFilePointer();
                else
                    return fileMeta.end - in.getFilePointer();
            }

        }

        @Override
        public int available() throws IOException {
            return (int) remaining();
        }

        @Override
        public void close() throws IOException {
        }

    }

    public static void main(String[] args) throws Exception {
        LogSystem logSystem = new LogSystem("logDisk", 125);
        logSystem.format();

        byte[] data = new byte[8];
        ByteBuffer b = ByteBuffer.wrap(data);

        OutputStream o = logSystem.newLog("log1");
        for (long i = 0; i < 10; i++) {
            b.rewind();
            b.putLong(i);
            o.write(data);
        }

        printAll(logSystem);


        o = logSystem.newLog("log2");
        for (long i = 0; i < 5; i++) {
            b.rewind();
            b.putLong(i + 10);
            o.write(data);
        }

        printAll(logSystem);



        logSystem.remove("log1");
        o = logSystem.newLog("log3");
        for (long i = 0; i < 10; i++) {
            b.rewind();
            b.putLong(i);
            o.write(data);
        }

        printAll(logSystem);  
    }

    private static void printAll(LogSystem logSystem) throws Exception {
        byte[] data = new byte[8];
        ByteBuffer b = ByteBuffer.wrap(data);
        for (String name : logSystem.listFiles()) {
            System.out.println(name);
            InputStream in = logSystem.readLog(name);
            while (in.available() > 0) {
                int bytes = in.read(data);
                if (bytes != 8)
                    throw new RuntimeException("wrong " + bytes);
                b.rewind();
                System.out.println(b.getLong());
            }
        }
    }
}
